package avicit.bdp.dcs.datasource.service.ds;

import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dcs.plugin.hbase.HBaseDS;
import avicit.bdp.dcs.utils.JdbcConstants;
import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * produce ds in this custom defined ds factory.
 */
public class DSFactory {

    private static final Logger logger = LoggerFactory.getLogger(DSFactory.class);

    /**
     * getDatasource
     *
     * @param dbType    dbType
     * @param parameter parameter
     * @return getDatasource
     */
    public static BaseDS getDatasource(String dbType, String parameter) {
        try {
            switch (dbType) {
                case JdbcConstants.MYSQL:
                    return JSON.parseObject(parameter, MySQLDS.class);
                case JdbcConstants.POSTGRESQL:
                    return JSON.parseObject(parameter, PostgreDS.class);
                case JdbcConstants.HIVE:
                    return JSON.parseObject(parameter, HiveDS.class);
                case JdbcConstants.ORACLE:
                    return JSON.parseObject(parameter, OracleDS.class);
                case JdbcConstants.GREENPLUM:
                    return JSON.parseObject(parameter, GreenPlumDS.class);
                case JdbcConstants.HDFS:
                    return JSON.parseObject(parameter, HdfsDS.class);
                case JdbcConstants.KAFKA:
                    return JSON.parseObject(parameter, KafkaDS.class);
                case JdbcConstants.IOTDB:
                    return JSON.parseObject(parameter, IoTDBDS.class);
                case JdbcConstants.HBASE:
                    return JSON.parseObject(parameter, HBaseDS.class);
                default:
                    return null;
            }
        } catch (Exception e) {
            logger.error("get ds object error", e);
            return null;
        }
    }

    /**
     * getDatasource
     *
     * @param dataSource dataSource
     * @return getDatasource
     */
    public static BaseDS getDatasource(DataSourceDTO dataSource) {
        String dbType = dataSource.getDatasourceType().toLowerCase();
        String parameter = dataSource.getConnectionParams();
        BaseDS baseDataSource = null;
        try {
            switch (dbType) {
                case JdbcConstants.MYSQL:
                    baseDataSource = JSON.parseObject(parameter, MySQLDS.class);
                    break;
                case JdbcConstants.POSTGRESQL:
                    baseDataSource = JSON.parseObject(parameter, PostgreDS.class);
                    break;
                case JdbcConstants.HIVE:
                    baseDataSource = JSON.parseObject(parameter, HiveDS.class);
                    break;
                case JdbcConstants.ORACLE:
                    baseDataSource = JSON.parseObject(parameter, OracleDS.class);
                    break;
                case JdbcConstants.GREENPLUM:
                    baseDataSource = JSON.parseObject(parameter, GreenPlumDS.class);
                    break;
                case JdbcConstants.HDFS:
                    baseDataSource = JSON.parseObject(parameter, HdfsDS.class);
                    break;
                case JdbcConstants.KAFKA:
                    baseDataSource = JSON.parseObject(parameter, KafkaDS.class);
                    break;
                case JdbcConstants.IOTDB:
                    baseDataSource = JSON.parseObject(parameter, IoTDBDS.class);
                    break;
                case JdbcConstants.OSS:
                    baseDataSource = JSON.parseObject(parameter, OssDS.class);
                    break;
                case JdbcConstants.HBASE:
                    baseDataSource = JSON.parseObject(parameter, HBaseDS.class);
                    break;
                case JdbcConstants.DM:
                    baseDataSource = JSON.parseObject(parameter, DMDS.class);
                    break;
                case JdbcConstants.API:
                    baseDataSource = JSON.parseObject(parameter, ApiDS.class);
                    break;
                case JdbcConstants.ODPS:
                    baseDataSource = JSON.parseObject(parameter, OdpsDS.class);
                    break;
                default:
                    baseDataSource = null;
            }

        } catch (Exception e) {
            logger.error("get dataSource object error", e);
        }
        return baseDataSource;
    }

    /**
     * load class
     *
     * @param dbType
     * @throws Exception
     */
    public static void loadClass(String dbType) throws Exception {
        switch (dbType) {
            case JdbcConstants.MYSQL:
                Class.forName(Constants.COM_MYSQL_JDBC_DRIVER);
                break;
            case JdbcConstants.POSTGRESQL:
                Class.forName(Constants.ORG_POSTGRESQL_DRIVER);
                break;
            case JdbcConstants.HIVE:
                Class.forName(Constants.ORG_APACHE_HIVE_JDBC_HIVE_DRIVER);
                break;
            case JdbcConstants.ORACLE:
                Class.forName(Constants.COM_ORACLE_JDBC_DRIVER);
                break;
            case JdbcConstants.GREENPLUM:
                Class.forName(Constants.COM_GREENPLUM_JDBC_DRIVER);
                break;
            case JdbcConstants.PRESTO:
                Class.forName(Constants.PRESTO_DRIVER);
                break;
            default:
                logger.error("not support sql type: {},can't load class", dbType);
                throw new IllegalArgumentException("not support sql type,can't load class");
        }
    }

}
